package rxgo

import (
	"context"
	"reflect"
	"fmt"
	"time"
	"sync"
)

type filteringOperator struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

/**
 * op
 *
 * @param o Observalbe
 * @param ctx context.Context
 *
 * @return 
 */
func (ftop filteringOperator) op(ctx context.Context, o *Observable) {
	in := o.pred.outflow
	out := o.outflow
	var wg sync.WaitGroup
	go func() {
		end := false
		for x := range in {
			if end {
				break
			}
			xv := reflect.ValueOf(x)
			if e, ok := x.(error); ok && !o.flip_accept_error {
				o.sendToFlow(ctx, e, out)
				continue
			}
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if ftop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				go func() {
					defer wg.Done()
					if ftop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}
		if o.flip != nil {
			buffer := (reflect.ValueOf(o.flip))
			if buffer.Kind() != reflect.Slice {
				panic("flip is not buffer")
			}
			for i := 0; i < buffer.Len(); i++ {
				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}
		wg.Wait()
		o.closeFlow(out)
	}()
}

/**
 * Debounce
 *
 * @param timespan time.Duration
 *
 * @return o Observalbe
 */
func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) {
	o = parent.newFilterObservable("debounce")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count := 0
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			count++
			go func() {
				tempCount := count
				time.Sleep(timespan)
				select {
				case <-ctx.Done():
					return
				default:
					if tempCount == count {
						o.sendToFlow(ctx, item.Interface(), out)
					}
				}
			}()
			return false
		},
	}
	return o
}

/**
 * Distinct
 *
 * @param
 *
 * @return o Observalbe
 */
func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	m := map[string]bool{}
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			itemStr := fmt.Sprintf("%v", item)
			if _, ok := m[itemStr]; !ok {
				m[itemStr] = true
				o.sendToFlow(ctx, item.Interface(), out)
			}
			return false
		},
	}
	return o
}


/**
 * ElementAt
 *
 * @param num int
 *
 * @return o Observalbe
 */
func (parent *Observable) ElementAt(num int) (o *Observable) {
	o = parent.newFilterObservable("elementAt.n")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count := 0
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			if count == num {
				o.sendToFlow(ctx, item.Interface(), out)
				return true
			}
			count++
			return false
		},
	}

	return o
}

/**
 * First
 *
 * @param 
 *
 * @return o Observalbe
 */
func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("first")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			o.sendToFlow(ctx, item.Interface(), out)
			return true
		},
	}
	return o
}

/**
 * IgnoreElements
 *
 * @param 
 *
 * @return o Observalbe
 */
func (parent *Observable) IgnoreElements() (o *Observable) {
	o = parent.newFilterObservable("ignoreElements")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			return false
		},
	}
	return o
}

/**
 * Last
 * @param 
 *
 * @return o Observalbe
 */
func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("last")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			o.flip = append([]interface{}{}, item.Interface())
			return false
		},
	}
	return o
}

/**
 * Sample
 *
 * @param sample chan interface{}
 *
 * @return o Observalbe
 */
func (parent *Observable) Sample(sample chan interface{}) (o *Observable) {
	o = parent.newFilterObservable("sample")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	var latest interface{} = nil
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			latest = item.Interface()
			go func() {
				tempEnd := true
				for tempEnd {
					select {
					case <-ctx.Done():
						tempEnd = true
					case <-sample:
						if latest != nil {
							if o.sendToFlow(ctx, latest, out) {
								tempEnd = false
							}
							latest = nil
						}
					}
				}
			}()
			return false
		},
	}
	return o
}

/**
 * Skip
 *
 * @param num int
 *
 * @return o Observalbe
 */
func (parent *Observable) Skip(num int) (o *Observable) {
	o = parent.newFilterObservable("skip.n")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count := 0
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			count++
			if count > num {
				o.sendToFlow(ctx, item.Interface(), out)
			}
			return false
		},
	}

	return o
}

/**
 * SkipLast
 *
 * @param num int
 *
 * @return o Observalbe
 */
func (parent *Observable) SkipLast(num int) (o *Observable) {
	o = parent.newFilterObservable("skipLast.n")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count := 0
	var lasts []interface{}
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			if count == num {
				o.sendToFlow(ctx, lasts[0], out)
				lasts = lasts[1:]
			} else {
				count++
			}
			lasts = append(lasts, item.Interface())
			return false
		},
	}

	return o
}

/**
 * Take
 */
func (parent *Observable) Take(num int) (o *Observable) {
	o = parent.newFilterObservable("take.n")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count := 0
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			count++
			if count > num {
				return true
			}
			o.sendToFlow(ctx, item.Interface(), out)
			return false
		},
	}

	return o
}

/**
 * TakeLast
 */
func (parent *Observable) TakeLast(num int) (o *Observable) {
	o = parent.newFilterObservable("takeLast.n")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	count := 0
	var lasts []interface{}
	o.operator = filteringOperator{
		opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
			count++
			if count <= num {
				lasts = append(lasts, item.Interface())
			} else {
				lasts = lasts[1:]
				lasts = append(lasts, item.Interface())
			}
			o.flip = lasts
			return false
		},
	}

	return o
}

/**
 * newFilterObservable
 */
func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	o = newObservable()
	o.Name = name

	parent.next = o
	o.pred = parent
	o.root = parent.root

	o.buf_len = BufferLen
	return o
}

